package club.monkeywood.ad.dmp.graphx

import org.apache.log4j.{Level, Logger}
import org.apache.spark.graphx.{Edge, Graph, VertexId, VertexRDD}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 查询有共同好友的人群
  * 使用Spark 连通图计算测试demo
  */
object ComFriends {


  /**
    * @author Mr.zhao
    *         2018/6/4 23:11
    */

  Logger.getLogger("org").setLevel(Level.DEBUG)

  def main(args: Array[String]): Unit = {

    // 1 创建sparkconf->SparkSession
    val sparkConf = new SparkConf()
    sparkConf.setAppName(s"${this.getClass.getSimpleName}")
    sparkConf.setMaster("local[*]")
    //使用KryoSerializer更快
    // RDD 序列化到磁盘 worker与worker之间的数据传输
    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    //避免字段名过长报错
    sparkConf.set("spark.debug.maxToStringFields", "100")
    val ss = SparkSession.builder()
      .appName(s"${this.getClass.getSimpleName}")
      .config(sparkConf)
      .getOrCreate()

    // 点集合 RDD[(Long, String)]
    // 点集合元素：(顶点Id, (用户姓名,年龄))
    val vertexRDD: RDD[(VertexId, (String, Int))] = ss.sparkContext.parallelize(Array(

      (1L, ("张无忌", 16)),
      (2L, ("赵敏", 17)),
      (9L, ("阿珠", 19)),
      (6L, ("张翠山", 20)),
      (133L, ("张三丰", 23)),

      (138L, ("金毛狮王", 24)),
      (16L, ("白眉鹰王", 25)),
      (44L, ("周芷若", 22)),
      (21L, ("殷素素", 14)),

      (158L, ("杨逍", 15)),
      (7L, ("杨不悔", 16)),
      (5L, ("纪晓芙", 30))
    ))
    // 边集合
    // 边集合元素：(定点1,定点2,方向)
    // 定点1和定点2联通
    // 边上无方向
    val edgesRDD: RDD[Edge[Int]] = ss.sparkContext.parallelize(Array(
      Edge(1, 133, 0),
      Edge(2, 133, 0),
      Edge(9, 133, 0),
      Edge(6, 133, 0),

      Edge(6, 138, 0),
      Edge(21, 138, 0),
      Edge(16, 138, 0),
      Edge(44, 138, 0),

      Edge(5, 158, 0),
      Edge(7, 158, 0)
    ))
    /**
      * 构建图:((姓名,年龄), 共同的顶点id)
      */
    val graph: Graph[(String, Int), Int] = Graph(vertexRDD,edgesRDD)
    /**
      * 查询连通图
      * connectedComponents 可以找到图中可以联通图的分支    2个分支
      * 连通图：通过边连接在一起的顶点
      * commonV元素：(userId,共通的顶点id)
      * 共通的顶点：取连通图中id最小的顶点
      */
    val commonV: VertexRDD[VertexId] = graph.connectedComponents().vertices
    commonV.foreach(println)
    //查询共同好友数组
    //第一组：(1,List(16, 1, 9, 138, 2, 44, 21, 133, 6))
    //第二组：(5,List(5, 158, 7))
    commonV.map(tp=>(tp._2, List(tp._1)))   //(共通的顶点id, List(userid))
        .reduceByKey(_++_)   //List(userid1)++List(userid2)=List(userid1, userid2)
        .foreach(println)

    /**
      * 查询共同好友组的姓名和年龄
      * vertexRDD元素：(userId, (姓名, 年龄))
      * commonV元素：(userId,共通的顶点id)
      */
    vertexRDD.join(commonV)
      //用偏函数匹配输入数据，转换数据格式
      .map {
        case (userId, ((name, age), minId)) => (minId, List((userId, name)))
      }
      .reduceByKey(_++_)
      .foreach(println)

    ss.stop()

  }


}
